package com.lagou.boot;

import com.lagou.client.NewRPCConsumer;
import com.lagou.protocol.RpcRequest;
import com.lagou.service.IUserService;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.log4j.MDC;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;

public class ConsumerBoot {

    //参数定义
    private static final String PROVIDER_NAME = "UserService#sayHello#";

    public static final Map<Integer, ZookeeperServer> zookeeperServer2 = new ConcurrentHashMap<Integer, ZookeeperServer>();

    static String  zookeeperpath="/provide_path";
    public static void main(String[] args) throws Exception {
        //获取连接
        CuratorFramework client = getCuratorClient();

        zkWatch(client,zookeeperpath);
    }

    private static CuratorFramework getCuratorClient() {
        RetryPolicy exponentialBackoffRetry = new ExponentialBackoffRetry(1000, 3);
        //使用fluent变成风格
        CuratorFramework curatorClient = CuratorFrameworkFactory.builder()
                .connectString("119.3.173.114:2181")
                .sessionTimeoutMs(1000)
                .connectionTimeoutMs(3000)
                .retryPolicy(exponentialBackoffRetry)
                .namespace("base")  //独立的命名空间
                .build();
        curatorClient.start();
        System.out.println("会话建立成功");
        return curatorClient;
    }


    /**
     * 启动服务
     */
    public static void runServer() throws InterruptedException {

//        //1.创建代理对象
        AtomicReference<IUserService> iUserService = new AtomicReference<>();
        final Long[] time = {0L};
        zookeeperServer2.forEach((k, v) -> {
            if (System.currentTimeMillis() - v.getLastTime() > 5000){
                return;
            }
            if (v.getLastTime() > time[0]){
                iUserService.set(v.getiUserService());
                time[0] = v.getLastTime();
            }
        });

        //2.循环给服务器写数据
        while (iUserService.get() != null){
            RpcRequest rpcRequest = new RpcRequest();
            rpcRequest.setClassName("com.lagou.protocol.RpcRequest");
            rpcRequest.setMethodName("getResult");
            rpcRequest.setRequestId("userServiceImpl");
            rpcRequest.setParameters(new Object[]{"Hi Server"});
            rpcRequest.setParameterTypes(new Class[]{String.class});
            String result = iUserService.get().getResult(rpcRequest);
            System.out.println(result);
            Thread.sleep(2000);
        }
    }
    /**
     * 注册监听
     * TreeCache: 可以将指定的路径节点作为根节点（祖先节点），对其所有的子节点操作进行监听，
     * 呈现树形目录的监听，可以设置监听深度，最大监听深度为 int 类型的最大值。
     */
    private static void zkWatch(final CuratorFramework client, final String path) throws Exception {
        TreeCache treeCache = new TreeCache(client, path);

        treeCache.getListenable().addListener(new TreeCacheListener() {
            public void childEvent(CuratorFramework client, TreeCacheEvent event) {
                ChildData eventData = event.getData();
                switch (event.getType()) {
                    case NODE_ADDED:
                        System.out.println(path + "节点添加" + eventData.getPath() + "\t添加数据为：" + new String(eventData.getData()));
                        String path1 = eventData.getPath();
                        String time = new String(eventData.getData());
                        String[] nodaNames = path1.split(":");
                        final int port = Integer.parseInt(nodaNames[1]);
                        //创建代理对象
                        IUserService service = (IUserService) NewRPCConsumer.createProxy(IUserService.class,nodaNames[0],port, zookeeperServer2);
                        if (zookeeperServer2.get(port) == null){
                            ZookeeperServer mapper = new ZookeeperServer();
                            mapper.setiUserService(service);
                            mapper.setLastTime(Long.valueOf(time));
                            zookeeperServer2.put(port, mapper);
                        }
                        /*Thread thread = new Thread(new Runnable() {
                            public void run() {
                                while (ZookeeperServer.get(port) != null){
                                    runServer(port);
                                    try {
                                        Thread.sleep(3000);
                                    } catch (InterruptedException e) {
                                        e.printStackTrace();
                                    }
                                }
                            }
                        });
                        thread.start();*/
                        break;
                    case NODE_UPDATED:
                        System.out.println(eventData.getPath() + "节点数据更新\t更新数据为：" + new String(eventData.getData()) + "\t版本为：" + eventData.getStat().getVersion());
                        String path = eventData.getPath();
                        String time1 = new String(eventData.getData());
                        final int port1 = Integer.parseInt(path.substring(path.lastIndexOf("/") + 1));
                        zookeeperServer2.get(port1).setLastTime(Long.valueOf(time1));
                        break;
                    case NODE_REMOVED:
                        System.out.println(eventData.getPath() + "节点被删除");
                        String path2 = eventData.getPath();
                        int i = Integer.parseInt(path2.substring(path2.lastIndexOf("/") + 1));
                        ConsumerBoot.zookeeperServer2.remove(i);
                        break;
                    default:
                        break;
                }
            }
        });
        treeCache.start();

        while (true){
            runServer();
            Thread.sleep(3000);
        }

//        Thread.sleep(Integer.MAX_VALUE);
    }
}
